package cn.doitedu.cn.doitedu.rtdw.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/8/31
 * @Desc: 学大数据，上多易教育
 *    截止到此刻，当天总的pv数，uv数，并且每 5分钟更新一次结果
 *
 *
 *
 *
 **/
public class Job2_TotalPvUvSe_Per5min {

    public static void main(String[] args) {

        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射kafka中的dwd层的行为日志明细
        tenv.executeSql(
                "  CREATE TABLE dwd_events_kafka(                   "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     user_id           BIGINT,                     "
                        + "     rt AS to_timestamp_ltz(event_time,3),         " // 表达式字段
                        + "     WATERMARK FOR  rt AS  rt -  INTERVAL '0' SECOND  "  // 声明watermark，目标字段就具备了时间属性，不再是一个普通字段
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd-events-detail',                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',         "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           ");


        // 建表，映射mysql中的结果表
        tenv.executeSql(
                " create table dash_total_pv_uv_sv_mysql(     " +
                        "    window_start timestamp(3),            " +
                        "    window_end timestamp(3),              " +
                        "    pv_cnt bigint,                        " +
                        "    session_cnt bigint,                   " +
                        "    uv_cnt bigint,                        " +
                        "    PRIMARY KEY (window_start,window_end) NOT ENFORCED     " +
                        " ) with (                                      " +
                        "  'connector' = 'jdbc',                            " +
                        "  'url' = 'jdbc:mysql://doitedu:3306/doit40',  " +
                        "  'table-name' = 'dash_total_pv_uv_sv',             " +
                        "  'username' = 'root',                             " +
                        "  'password' = 'root'                              " +
                        " )                                                 "
        );

        // 运算sql
        tenv.executeSql(
                " insert into dash_total_pv_uv_sv_mysql  select                                                  "+
                        "     window_start,                                                                                "+
                        "  	window_end,                                                                                    "+
                        "  	sum(if(event_id='page_load',1,0)) as pv_cnt,                                                   "+
                        "  	count(distinct session_id) as session_cnt,                                                     "+
                        "  	count(distinct user_id) as uv_cnt                                                              "+
                        "  from TABLE(                                                                                     "+
                        "     CUMULATE(TABLE dwd_events_kafka,DESCRIPTOR(rt), INTERVAL '5' MINUTE ,INTERVAL '24' HOUR)     "+
                        "  )                                                                                               "+
                        "  GROUP BY                                                                                        "+
                        "     window_start,                                                                                "+
                        "  	window_end                                                                                     "
        );





    }


}
